package com.exmaple.mapreducer.writable;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

//map阶段
public class SalaryTotalMapper extends Mapper<LongWritable, Text, IntWritable,Employee> {

    private  IntWritable k2 = new IntWritable();//部门编号作为k2
    private   Employee v2 = new Employee();//员工的对象作为v2
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //1. v1由Text类型转换为String：toString()方法
          // 7369,SMITH,CLERK,7902,1980/12/17,800,,20
        String line = value.toString();
        //2. 按逗号进行分词：split(“,”)方法
        String[] datas = line.split(",");
        // 3. 取出需要的字段：部门号：字段[7]、工资：字段[5]
        //字段名 EMPNO, ENAME,    JOB,   MGR,   HIREDATE,  SAL, COMM, DEPTNO
        //数据类型：Int，Char，   Char  ， Int，  Date  ，   Int   Int，  Int
        //数据: 7654, MARTIN, SALESMAN, 7698, 1981/9/28, 1250, 1400, 30
        v2.setEmpno(Integer.parseInt(datas[0]));
        v2.setEname(datas[1]);
        v2.setJob(datas[2]);
        try {
            v2.setMgr(Integer.parseInt(datas[3]));//部门经理
        }catch (Exception e){
            //由于通过数据文件看到部门经理有可能为空,为空会报异常，在此当经理id为空的时候
            //就设置为一个默认值
            v2.setMgr(-1);
        }
        v2.setHiredate(datas[4]);
        v2.setSal(Integer.parseInt(datas[5]));//工资
        try {
            v2.setComm(Integer.parseInt(datas[6]));
        }catch (Exception e){
            //由于通过数据文件看到奖金有可能为空,为空会报异常，在此当奖金为空的时候
            //就设置为一个默认值
            v2.setComm(0);
        }
        v2.setDeptno(Integer.parseInt(datas[7]));

        // 3. 取出需要的字段：部门号：字段[7]、工资：字段[5]
//        String deptNo = datas[7]; //拿到部门编号
//        k2.set(Integer.parseInt(deptNo));
        //v2.set(Integer.parseInt(datas[5]));
        k2.set(v2.getDeptno());

        // 4. 输出k2, v2：context.write()方法
        // context.write(k2,v2);
        context.write(k2,v2);






    }
}
